package com.doitedu.core2

import org.apache.spark.util.AccumulatorV2

import scala.collection.mutable

/**
 * @Date: 22.7.4 
 * @Author: HANGGE
 * @qq: 598196583
 * @Tips: 学大数据 ,到多易教育
 * @Description:
 * 统计单词的个数
 * 输入单词
 * Map[String ,Int]
 *
 *
 */
class C07_自定义累加器 extends  AccumulatorV2[String ,mutable.Map[String,Int]] {

  val mp = mutable.Map[String, Int]()

  /**
   * 判断累加是是否是初始状态
   * @return
   */
  override def isZero: Boolean = mp.isEmpty

  override def copy(): AccumulatorV2[String, mutable.Map[String, Int]] =new  C07_自定义累加器()

  /**
   * 重置 将返回的值变成初始值
   *   累加器的初始值  0
   */
  override def reset(): Unit = mp.clear()


  /**
   * 在每个Task上的计算逻辑
   *  输入单词   统计这个单词在这个分区内的个数
   * @param v
   * a  b  a  b c
   */
  override def add(word: String): Unit = {
    var cnt = mp.getOrElse(word, 0)
    cnt +=1
    mp.put(word , cnt)
  }

  /**
   * 各个Task数据的合并  返回最终的结果数据
   * 每个单词  总次数
   * @param other
   * 这个方法聚合方法
   * 有多少个分区调用计策  参数是每个分区的结果[Map]
   */
  override def merge(other: AccumulatorV2[String, mutable.Map[String, Int]]): Unit = {
    // 当前Map  和  另一个Task的Map合并获取结果   reduce
    // 将各个分区的Map数据合并到mp集合中
    // 累加器开始是空的   []
    //  [a,2]  [b,3]  task01                 mp
    //  [a,2]  [b,3]  [c,5]  task02
    //  [a,2] [c,5] [d,5] task03

    val taskMap = other.value
    // 将每个分区的素具合并到返回的结果中  mp
    taskMap.foreach(tp=>{
      val word = tp._1
      val cnt = tp._2
      var i = mp.getOrElse(word, 0)
      i+=cnt
     mp.put(word ,i)
    })
  }

  /**
   * 返回结果
   * @return
   */
  override def value: mutable.Map[String, Int] =  mp

}
